package cn.ant.mqHandle;

import cn.ant.config.RabbitMQConfig;
import cn.ant.entity.MessageSendDTO;
import com.alibaba.fastjson2.JSONObject;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/***
 * 生产者代码
 * @author Anhui AntLaddie <a href="https://juejin.cn/user/4092805620185316">(掘金蚂蚁小哥)</a>
 * @version 1.0
 **/
@Slf4j
@Component
@RequiredArgsConstructor
public class ProducerSend {

    // 注入rabbitTemplate对象
    private final RabbitTemplate rabbitTemplate;
    // 注入StringRedisTemplate对象
    private final StringRedisTemplate redisTemplate;

    /***
     * 生产者方法
     * @param msg 需要投递的消息
     */
    public void producerSendMsg(MessageSendDTO msg) {
        // Redis Key
        String key = "rabbit:confirm:order:" + msg.getMsgID();
        // redis的键值操作对象
        ValueOperations<String, String> forValue = redisTemplate.opsForValue();
        try {
            // 防止重复提交
            // 若设置成功则为True，设置不成功或者设置的值已经存在则返回False
            // 这里设置20秒代表自动过期，一旦设置这个键值，消息被成功投递则不删除（防止20秒内重复提交），但是投递失败
            // 以后我需要删除这个键值，方便下次继续设置投递；；具体按照实际设置过期时间
            Boolean deliver = forValue.setIfAbsent(key, String.valueOf(msg.getMsgID()), 20, TimeUnit.SECONDS);

            // 判断设置成功则发送消息（否则这个消息可能多次发送给消费者）
            if (Boolean.TRUE.equals(deliver)) {
                // 延迟测试（假设投递花了5秒），可以看到重复的问题演示
                Thread.sleep(5000);
                // 消息转换为JSON格式并转为字节数组
                byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8);
                // 设置回退的相关信息，并设置回调信息（里面包含交换机和路由key，方便后面在回调的时候重新投递）
                CorrelationData cd = new CorrelationData();
                cd.setReturned(new ReturnedMessage(new Message(bytes), 0, null,
                        RabbitMQConfig.ORDINARY_DIRECT_EXCHANGE, RabbitMQConfig.ROUTE_KEY));
                // 发送消息
                rabbitTemplate.convertAndSend(RabbitMQConfig.ORDINARY_DIRECT_EXCHANGE,
                        RabbitMQConfig.ROUTE_KEY, bytes, cd);
            } else {
                log.info("消息已经由生产者发送投递了，请忽重复投递！");
            }
        } catch (Exception e) {
            // 若生产者投递出现问题则代表投递不成功，删除这次缓存
            redisTemplate.delete(key);
        }
    }
}
